package com.ds.infra.rpc.core.transport.client.handler;

import com.ds.infra.rpc.core.constant.RpcConstants;
import com.ds.infra.rpc.core.protocol.RpcProtocol;
import com.ds.infra.rpc.core.protocol.RpcRequest;
import com.ds.infra.rpc.core.transport.client.NettyClient;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 客户端连接管理器（含心跳检测/断线重连）
 *
 * @author WeiShaoying
 */
@Slf4j
public class RpcClientHandler extends ChannelInboundHandlerAdapter {

    private final NettyClient nettyClient;
    private final ScheduledExecutorService reconnectExecutor;
    private volatile boolean isActive = false;

    public RpcClientHandler(NettyClient nettyClient) {
        this.nettyClient = nettyClient;
        this.reconnectExecutor = Executors.newSingleThreadScheduledExecutor();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        this.isActive = true;
        log.info("Connection established to {}", ctx.channel().remoteAddress());
        // 启动心跳检测
        scheduleHeartbeat(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        handleDisconnection(ctx);
    }

    private void handleDisconnection(ChannelHandlerContext ctx) {
        if (isActive) {
            isActive = false;
            log.warn("Connection lost with {}", ctx.channel().remoteAddress());
            // 延迟重连（避免频繁重试）
            reconnectExecutor.schedule(() -> {
                log.info("Attempting to reconnect...");
                nettyClient.reconnect();
            }, 5, TimeUnit.SECONDS);
        }
    }

    private void scheduleHeartbeat(ChannelHandlerContext ctx) {
        ctx.executor().scheduleAtFixedRate(() -> {
            if (ctx.channel().isActive()) {
                RpcProtocol<RpcRequest> heartbeat = buildHeartbeat();
                ctx.writeAndFlush(heartbeat).addListener(future -> {
                    if (!future.isSuccess()) {
                        log.error("Heartbeat send failed", future.cause());
                    }
                });
            }
        }, 0, 30, TimeUnit.SECONDS);
    }

    private RpcProtocol<RpcRequest> buildHeartbeat() {
        RpcRequest request = new RpcRequest();
        request.setRequestId("HEARTBEAT_" + System.currentTimeMillis());

        return new RpcProtocol<>(
                new RpcProtocol.Header(
                        RpcConstants.MAGIC,
                        (byte) 1,
                        RpcProtocol.MsgType.HEARTBEAT.getValue(),
                        0 // 长度占位
                ),
                request
        );
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("Client connection error", cause);
        handleDisconnection(ctx);
        ctx.close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 处理服务端PONG响应
        if (msg instanceof RpcProtocol && ((RpcProtocol<?>) msg).getHeader().getMsgType()
                == RpcProtocol.MsgType.HEARTBEAT.getValue()) {
            log.debug("Received heartbeat ack");
        } else {
            ctx.fireChannelRead(msg); // 传递给下一个handler
        }
    }

    public void shutdown() {
        reconnectExecutor.shutdownNow();
    }
}
